package org.mq.mymq.nameserver.event.listener;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import io.netty.util.internal.StringUtil;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.enums.NameServerResponseCode;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.event.model.UnRegistryEvent;
import org.mq.mymq.nameserver.store.ServiceInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @ClassName UnRegistryListener
 * @Description 下线事件监听器
 * @Author jiarong_ye
 * @Date 2025/6/19 13:23
 * @Version 1.0
 */
public class UnRegistryListener implements Listener<UnRegistryEvent> {
    private static final Logger log = LoggerFactory.getLogger(UnRegistryListener.class);

    @Override
    public void onReceive(UnRegistryEvent event) throws IllegalAccessException {
        ChannelHandlerContext ctx = event.getChannelHandlerContext();
        Object reqId = ctx.attr(AttributeKey.valueOf("reqId")).get();
        TcpMsg tcpMsg = null;
        if (reqId == null) {
            tcpMsg = new TcpMsg(NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode(),
                    NameServerResponseCode.ERROR_USER_OR_PASSWORD.getDesc().getBytes());
            ctx.writeAndFlush(tcpMsg);
            ctx.close();
            throw new IllegalAccessException("please registry first");
        }
        // 下线监听器中识别master节点
        ServiceInstance serviceInstance = CommonCache.getServiceInstanceManager().get((String) reqId);
        Map<String, Object> attrs = serviceInstance.getAttrs();
        String brokerClusterRole = (String) attrs.getOrDefault("role", "");
        String brokerClusterGroup = (String) attrs.getOrDefault("group", "");
        // 单机架构不需要处理，直接发送下线通知并关闭通道
        boolean isClusterMode = !StringUtil.isNullOrEmpty(brokerClusterRole) && !StringUtil.isNullOrEmpty(brokerClusterGroup);
        if (isClusterMode) {
            // 集群模式主节点情况
            if ("master".equals(brokerClusterRole)) {
                // todo 主从切换逻辑处理
                log.error("master node fail");
                List<ServiceInstance> slaveList = new ArrayList<>();
                for (ServiceInstance insatnce : CommonCache.getServiceInstanceManager().getServiceInstanceMap().values()) {
                    String matchGroup = (String) insatnce.getAttrs().getOrDefault("group", "");
                    String matchRole = (String) insatnce.getAttrs().getOrDefault("role", "");
                    if (matchGroup.equals(brokerClusterGroup) && "slave".equals(matchRole)) {
                        slaveList.add(insatnce);
                    }
                }
                log.info("get same cluster group slave node:{}", JSON.toJSON(slaveList));
                // 移除之前的master节点，从节点中选择最新的一方作为主节点
                long maxVersion = 0L;
                ServiceInstance newMasterServiceInstance = null;
                for (ServiceInstance instance : slaveList) {
                    long latestVersion = (long) instance.getAttrs().getOrDefault("latestVersion", 0L);
                    if (maxVersion <= latestVersion) {
                        newMasterServiceInstance = instance;
                        maxVersion = latestVersion;
                    }
                }
                CommonCache.getServiceInstanceManager().remove(reqId.toString());
                if (newMasterServiceInstance != null) {
                    newMasterServiceInstance.getAttrs().put("role", "master");
                }
                // 重新设置主从关系
                CommonCache.getServiceInstanceManager().reload(slaveList);
                log.info("new cluster node is:{}", JSON.toJSONString(newMasterServiceInstance));
            }
            return;
        }
        CommonCache.getServiceInstanceManager().remove(reqId.toString());
        log.info("链接断开处理");

    }
}
